package com.atguigu.realtime.gmalllogger.controller;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;

/**
 * @Author lzc
 * @Date 2020/11/9 11:38
 */
/*@Controller
@ResponseBody*/
@RestController // 等价于
@Slf4j
public class LoggerController {

    @PostMapping("/applog")
    public String appLog(@RequestBody String logString) {
        // 1. 把数据落盘, 给离线需求用
        saveToDisk(logString);
        // 2. 把数据直接发送到kafka
        sendToKafka(logString);
        return "ok";
    }

    @Autowired
    KafkaTemplate<String, String> kafka;

    /**
     * 把日志数据发送到kafka
     *
     * @param logString
     */
    private void sendToKafka(String logString) {
        // 解析成一个json对象
        JSONObject obj = JSON.parseObject(logString);
        // 把事件日志和启动日志写入到不同的topic
        if (obj.getString("start") != null && obj.getString("start").length() > 0) {
            kafka.send("gmall_startup_topic", logString);
        } else {
            kafka.send("gmall_event_topic", logString);
        }
    }

    /**
     * 把日志写入到磁盘
     *
     * @param logString
     */
    private void saveToDisk(String logString) {
        log.info(logString);
    }
}
/*
get 请求
     用于查询数据
     数据在url中

post 请求
    用于提交数据
    数据更安全
    数据没有在url中, 在请求体中


kafka的分区数据如何定?
    spark-streaming从kafka消费数据,
    直连的模式:

    由spark-steaming并行能力来定

 */